package avicit.bdp.dds.server.master.runner;

import avicit.bdp.common.utils.SpringApplicationContext;
import avicit.bdp.dds.dispatch.enums.ExecutionStatus;
import avicit.bdp.dds.dispatch.model.TwoTuple;
import avicit.bdp.core.util.json.JSONUtils;
import avicit.bdp.dds.common.Constants;
import avicit.bdp.dds.dao.AlertDao;
import avicit.bdp.dds.dao.entity.ProcessInstance;
import avicit.bdp.dds.dao.entity.TaskInstance;
import avicit.bdp.dds.server.log.TaskLogDiscriminator;
import avicit.bdp.dds.server.master.config.MasterConfig;
import avicit.bdp.dds.service.process.ProcessService;
import avicit.bdp.dds.service.queue.TaskPriorityQueue;
import avicit.bdp.dds.service.queue.TaskPriorityQueueImpl;
import ch.qos.logback.classic.LoggerContext;
import ch.qos.logback.classic.sift.SiftingAppender;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.Callable;

import static avicit.bdp.dds.common.Constants.UNDERLINE;


/**
 * @author admin
 * master task exec base class
 */
public class MasterBaseTaskExecThread implements Callable<TwoTuple<Boolean, String>> {

    /**
     * logger of MasterBaseTaskExecThread
     */
    protected Logger logger = LoggerFactory.getLogger(getClass());


    /**
     * process service
     */
    protected ProcessService processService;

    /**
     * alert database access
     */
    protected AlertDao alertDao;

    /**
     * process instance
     */
    protected ProcessInstance processInstance;

    /**
     * task instance
     */
    protected TaskInstance taskInstance;

    /**
     * whether need cancel
     */
    protected boolean cancel;

    /**
     * master config
     */
    protected MasterConfig masterConfig;

    /**
     * taskUpdateQueue
     */
    private TaskPriorityQueue taskUpdateQueue;

    /**
     * constructor of MasterBaseTaskExecThread
     *
     * @param taskInstance task instance
     */
    public MasterBaseTaskExecThread(TaskInstance taskInstance) {
        this.processService = SpringApplicationContext.getBean(ProcessService.class);
        this.alertDao = SpringApplicationContext.getBean(AlertDao.class);
        this.cancel = false;
        this.taskInstance = taskInstance;
        this.masterConfig = SpringApplicationContext.getBean(MasterConfig.class);
        this.taskUpdateQueue = SpringApplicationContext.getBean(TaskPriorityQueueImpl.class);
    }

    /**
     * get task instance
     *
     * @return TaskInstance
     */
    public TaskInstance getTaskInstance() {
        return this.taskInstance;
    }

    /**
     * kill master base task exec thread
     */
    public void kill() {
        this.cancel = true;
    }

    /**
     * submit master base task exec thread
     *
     * @return TaskInstance
     */
    protected TaskInstance submit() {
        Integer commitRetryTimes = masterConfig.getMasterTaskCommitRetryTimes();
        Integer commitRetryInterval = masterConfig.getMasterTaskCommitInterval();

        int retryTimes = 1;
        boolean submitDB = false;
        boolean submitTask = false;
        TaskInstance task = null;
        while (retryTimes <= commitRetryTimes) {
            try {
                if (!submitDB) {
                    // submit task to db
                    task = processService.submitTask(taskInstance);
                    if (task != null && StringUtils.isNotEmpty(task.getId())) {
                        submitDB = true;
                    }
                }
                if (submitDB && !submitTask) {
                    // dispatch task
                    submitTask = dispatchTask(task);
                }
                if (submitDB && submitTask) {
                    return task;
                }
                if (!submitDB) {
                    logger.error("task commit to db failed , taskId {} has already retry {} times, " +
                        "please check the database", taskInstance.getId(), retryTimes);
                } else if (!submitTask) {
                    logger.error("task commit  failed , taskId {} has already retry {} times, " +
                        "please check", taskInstance.getId(), retryTimes);
                }
                Thread.sleep(commitRetryInterval);
            } catch (Exception e) {
                logger.error("task commit to mysql and dispatcht task failed", e);
            }
            retryTimes += 1;
        }
        return task;
    }


    /**
     * dispatcht task
     *
     * @param taskInstance taskInstance
     * @return whether submit task success
     */
    public Boolean dispatchTask(TaskInstance taskInstance) {

        try {
            if (taskInstance.isConditionsTask() || taskInstance.isDependTask() || taskInstance.isSubProcess()) {
                return true;
            }
            if (ExecutionStatus.of(taskInstance.getState()).typeIsFinished()) {
                logger.info("submit task , but task {} state {} is already  finished. ",
                    taskInstance.getName(), taskInstance.getState().toString());
                return true;
            }
            // task cannot submit when running
            if (taskInstance.getState() == ExecutionStatus.RUNNING_EXEUTION.getCode()) {
                logger.info("submit to task, but task {} state already be running. ", taskInstance.getName());
                return true;
            }

            logger.info("task ready to submit: {}", taskInstance);

            /**
             *  taskPriorityInfo
             */
            String taskPriorityInfo = buildTaskPriorityInfo(processInstance.getProcessInstancePriority(),
                processInstance.getId(),
                taskInstance.getTaskInstancePriority(),
                taskInstance.getId(),
                Constants.DEFAULT_WORKER_GROUP);
            taskUpdateQueue.put(taskPriorityInfo);
            logger.info(String.format("master submit success, task : %s", taskInstance.getName()));
            return true;
        } catch (Exception e) {
            logger.error("submit task  Exception: ", e);
            logger.error("task error : %s", JSONUtils.toJson(taskInstance));
            return false;
        }
    }


    /**
     * buildTaskPriorityInfo
     *
     * @param processInstancePriority processInstancePriority
     * @param processInstanceId       processInstanceId
     * @param taskInstancePriority    taskInstancePriority
     * @param taskInstanceId          taskInstanceId
     * @param workerGroup             workerGroup
     * @return TaskPriorityInfo
     */
    private String buildTaskPriorityInfo(int processInstancePriority,
                                         String processInstanceId,
                                         int taskInstancePriority,
                                         String taskInstanceId,
                                         String workerGroup) {
        return processInstancePriority +
            UNDERLINE +
            processInstanceId +
            UNDERLINE +
            taskInstancePriority +
            UNDERLINE +
            taskInstanceId +
            UNDERLINE +
            workerGroup;
    }

    /**
     * submit wait complete
     *
     * @return true
     */
    protected TwoTuple<Boolean, String> submitWaitComplete() {
        return new TwoTuple<>(true, null);
    }

    /**
     * call
     *
     * @return boolean
     * @throws Exception exception
     */
    @Override
    public TwoTuple<Boolean, String> call() throws Exception {
        this.processInstance = processService.findProcessInstanceById(taskInstance.getProcessInstanceId());
        return submitWaitComplete();
    }

    /**
     * get task log path
     *
     * @return log path
     */
    public String getTaskLogPath(TaskInstance task) {
        String logPath;
        try {
            String baseLog = ((TaskLogDiscriminator) ((SiftingAppender) ((LoggerContext) LoggerFactory.getILoggerFactory())
                .getLogger("ROOT")
                .getAppender("TASKLOGFILE"))
                .getDiscriminator()).getLogBase();
            if (baseLog.startsWith(Constants.SINGLE_SLASH)) {
                logPath = baseLog + Constants.SINGLE_SLASH +
                    task.getProcessDefinitionId() + Constants.SINGLE_SLASH +
                    task.getProcessInstanceId() + Constants.SINGLE_SLASH +
                    task.getId() + ".log";
            } else {
                logPath = System.getProperty("user.dir") + Constants.SINGLE_SLASH +
                    baseLog + Constants.SINGLE_SLASH +
                    task.getProcessDefinitionId() + Constants.SINGLE_SLASH +
                    task.getProcessInstanceId() + Constants.SINGLE_SLASH +
                    task.getId() + ".log";
            }
        } catch (Exception e) {
            logger.error("logger", e);
            logPath = "";
        }
        return logPath;
    }




}
